package com.st.jms.use;

import com.st.jms.kafka.ConsumerClient;
import com.st.jms.kafka.ProducerAKClient;
import com.st.jms.util.ReadProperties;

import java.util.ArrayList;
import java.util.List;

/**
 * @author wuxianwei on 15/11/23.
 */
public class MainClass {
  public static void main(String[] args) {
    //从配置文件获取参数
    String topic= ReadProperties.getPropDef("topic");
    String brokerList= ReadProperties.getPropDef("brokerList");
    Integer sendBatch =Integer.parseInt(ReadProperties.getPropDef("sendBatch"));
    String zkConnect = ReadProperties.getPropDef("zkConnect");
    String groupId= ReadProperties.getPropDef("groupId");
    String groupId1 = ReadProperties.getPropDefIfBlank("","1");

    //开启监听线程
    TestMessageHandler testMessageHandler1 = new TestMessageHandler();
    ConsumerClient consumerClient1 = new ConsumerClient(zkConnect,topic,groupId,testMessageHandler1);
    consumerClient1.start();

    TestMessageHandler testMessageHandler2 = new TestMessageHandler();
    ConsumerClient consumerClient2 = new ConsumerClient(zkConnect,topic,groupId,testMessageHandler2);
    consumerClient2.start();

    TestMessageHandler testMessageHandler3 = new TestMessageHandler();
    ConsumerClient consumerClient3 = new ConsumerClient(zkConnect,topic,groupId1,testMessageHandler3);
//    consumerClient3.start();

    //发送测试消息
    ProducerAKClient producer = new ProducerAKClient(brokerList,sendBatch);
    producer.sendMessage(topic,"messageContent中文123456!from 单独发送.");
    System.out.println(1);
    List<String> mesList = new ArrayList<>(10);
    for (int i = 1; i <= 10; i++) {
      mesList.add("messageContent中文123456!from 批量发送 batch:"+i);
    }
    producer.putMessages(topic, mesList);
    System.out.println(10);
  }

}
